package producer_test

import (
	"container/list"
	"fmt"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/IBM/sarama"
	"github.com/petermattis/goid"
	"strconv"
	"time"
)

func execIBMProducer(topList *list.List, producerModel *model.KafkaProducerModel) {
	brokerAdds := []string{"localhost:9091"}

	var config = sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Retry.Backoff = 500 * time.Millisecond
	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	config.Producer.Return.Successes = true //重要配置，尝试不配置生产者创建为nil

	producer, e := sarama.NewSyncProducer(brokerAdds, config)
	if e != nil {
		logger.Infof("Failed to create producer: %v", e)
	}
	defer util.ClosePC1Io(&util.KafkaPCIo{
		Producer: &util.KafkaProType{SyncPro: producer},
	})
	defer close(producerModel.StopChan[goid.Get()])
	// 每个生产者者先轮训后随机获取一个主题(当轮旋完所有主题之后保证所有主题都被消费者占用就开始随机)
	topicName := cusTopicName(topList, producerModel)
	i := 0
	for {
		select {
		case <-producerModel.StopChan[goid.Get()]:
			logger.Info("kafka ibm producer shutting down")
			deleteSignMap(producerModel.StopChan, goid.Get())
			return
		case <-time.After(2 * time.Second):
		}
		// 构造消息
		value := fmt.Sprintf("%d号生产者使用IBM生产器生产数据=%s", goid.Get(), strconv.Itoa(i))
		mess := &sarama.ProducerMessage{
			Topic: topicName,
			Value: sarama.StringEncoder(value), // 消息内容
		}
		_, _, err := producer.SendMessage(mess)
		if err != nil {
			return
		}
		logger.Infof("生产者：%d号生产者使用IBM生产器生产数据=%s，发送消息到主题%s",
			goid.Get(), strconv.Itoa(i), topicName)
		i++
	}
}
